package com.inspur.cloud.service.dataspace;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.inspur.cloud.configuration.AmbariConfig;
import com.inspur.cloud.entity.ambari.ClusterConfigEntity;
import com.inspur.cloud.entity.dataspace.special.RangerEntity;
import com.inspur.cloud.entity.dataspace.special.RangerPolicyItem;
import com.inspur.cloud.entity.dataspace.special.RangerResource;
import com.inspur.cloud.enums.PermissionEnum;
import com.inspur.cloud.exception.ApiException;
import com.inspur.cloud.util.Base64Util;
import com.inspur.cloud.util.SshConf;
import com.inspur.cloud.util.SshUtil;
import net.minidev.json.JSONValue;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpException;
import org.apache.commons.httpclient.HttpMethodBase;
import org.apache.commons.httpclient.methods.*;
import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.*;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.socket.ConnectionSocketFactory;
import org.apache.http.conn.socket.PlainConnectionSocketFactory;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;

@Service
public class RangerService {

	public static String RANGER_REST_URL_V2 = "service/public/v2/api/policy";
	private static final Logger logger = LoggerFactory.getLogger(RangerService.class);

	@Autowired
    private RangerRestService rangerRestService;
    @Autowired
	private AmbariConfig ambariConfig;
	@Autowired
    private ClusterConfigUtil clusterConfigUtil;


	/**
	 * 增加或更新权限时调用
	 * 只有权限不为空的时候采用这个方法
	 * @param id
	 * @return
	 * @throws HttpException
	 * @throws IOException
	 */
    public Map<String, String> doAddOrUpdateRequest(int id, JSONObject policyBody) throws HttpException, IOException {
        //id小于0，则post请求新增策略，id大于0，则put请求更新策略
        String requestType = "POST";
        String url = getRangeUrl() + "/" + RANGER_REST_URL_V2;
        if (id > 0) {
            requestType = "PUT";
            url += "/" + id;
        }
        logger.info("RangerService doAddOrUpdateRequest id[" + id + "] url[" + url + "]");
        logger.info("RangerService doAddOrUpdateRequest param:" + policyBody.toString());
        Map<String, String> mapResult = doRangerRequest(requestType, url, policyBody.toJSONString());
		if(mapResult.get("statuscode").equals("200")){
			JSONObject jsonResponse = JSONObject.parseObject(mapResult.get("response"));
	        String policyid = jsonResponse.getString("id");
	        mapResult.put("policyid", policyid);
			logger.info("do request success");
		}else{
			throw new HttpException("statuscode["+mapResult.get("statuscode")+"], response[" + mapResult.get("response")+ "]");
		}

		return mapResult;
	}

	/**
	 * 标准增加ranger权限方法
	 * @param rangerEntity
	 */
	public Map<String, String> addRangerEntity(RangerEntity rangerEntity) {
		Map<String, String> result = new HashMap();

        //RangerEntity中必不可少的三项设置
        Map<String, String> rangerParams = new HashMap<>();
        List<RangerPolicyItem> rangerPolicyItems;
        List<RangerResource> rangerResources;
		try {
			String type = rangerEntity.getType();
			String resource = rangerEntity.getResource();
            String serviceName = rangerRestService.getServiceName(type);

			rangerResources = rangerEntity.getRangerResourceList();
			rangerPolicyItems = rangerEntity.getRangerPolicyItemList();

            rangerParams.put("isEnabled", rangerEntity.isEnabled() ? "true" : "false");
            rangerParams.put("isAuditEnabled", rangerEntity.isAuditEnabled() ? "true" : "false");
            rangerParams.put("service", serviceName);
            rangerParams.put("name", rangerEntity.getPolicyName());
            rangerParams.put("description", rangerEntity.getDescription());

            //判断策略是否已经存在，－1代表不存在
			int policyId = newSmartAdd(type,resource);
            //组装policyBody的JSON串
            JSONObject policyBody = buildJsonObject(policyId, rangerParams, rangerPolicyItems, rangerResources);
            //发起请求
            result = doAddOrUpdateRequest(policyId, policyBody);

			String res = result.get("statuscode");
			if (res.equals("200")) {
				logger.info("RangerService add policy for "+ policyId +" successfully.");
			}else {
				logger.warn("RangerService add policy for "+ policyId +" error.");
			}
		}catch (Exception e) {
			logger.error(e.getMessage());
            throw new ApiException("RangerService addRangerEntity error.");
		}
		return result;
	}


	/**
	 * 查询是否已经存在policyId了
	 * @param type
	 * @param resource
	 * @return
	 */
	public int newSmartAdd(String type,String resource){
        int policyId = rangerRestService.getPolicyIdByResourceNameAndPolicyName(type, resource);
        logger.info("RangerService newSmartAdd check type[" + type + "] resource[" + resource + "] result is {" + policyId + "}");
		return policyId;
	}
	/**
	 * 增加组合策略
	 * 如增加hive权限时，同时增加对应hdfs目录的权限
	 * @param type
	 * @param resource
	 * @param permissions 权限字符串，使用"|"分隔
	 * @param users 用户字符串，使用"|"分隔
	 */
	public void setPermissionInRanger(String type,String resource,String permissions,String users) {
		//关联hive的权限，hive数据库增加权限时，增加对应hdfs路径的权限
		if (type.equalsIgnoreCase("hive")) {
			setSinglePolicyPermissionInRanger(type, resource, permissions, users);
			//warehouse地址前缀
			String hiveMetaStoreDir = clusterConfigUtil.getHiveConfig().get("hive.metastore.warehouse.dir").toString();
			if (hiveMetaStoreDir.endsWith("/")) {
				hiveMetaStoreDir = hiveMetaStoreDir.substring(0, hiveMetaStoreDir.length() - 1);
			}
			String hiveMetaStoreExternalDir = clusterConfigUtil.getHiveConfig().get("hive.metastore.warehouse.external.dir").toString();
			if (hiveMetaStoreExternalDir.endsWith("/")) {
				hiveMetaStoreExternalDir = hiveMetaStoreExternalDir.substring(0, hiveMetaStoreExternalDir.length() - 1);
			}
			String[] resourceArray = resource.split("\\|");
			//如果权限设置的是列，则不增加对应hdfs权限，因没有对应的目录
			if (!resourceArray[2].equals("*")) return;
			//拼接库表
			String databaseAndTablePath = "";
			if (!resourceArray[0].equals("*")) {
				databaseAndTablePath = "/" + resourceArray[0] + ".db";
				if (!resourceArray[1].equals("*")) {
					databaseAndTablePath += "/" + resourceArray[1];
				}
			}
			//拼接对应hdfs路径
			String hiveMetaStoreOnHdfsDir = hiveMetaStoreDir + databaseAndTablePath;
			logger.info("RangerService setPermissionInRanger type:["+ type +"] and metastore dir:["+ hiveMetaStoreOnHdfsDir +"]");
			String hiveMetaStoreExternalOnHdfsDir = hiveMetaStoreExternalDir + databaseAndTablePath;
			logger.info("RangerService setPermissionInRanger type:["+ type +"] and external metastore dir:["+ hiveMetaStoreExternalOnHdfsDir +"]");
			//增加什么权限，暂时最大权限
			String hdfsPathPermissions = "";
			if (!(StringUtils.isEmpty(permissions))) {
				hdfsPathPermissions = PermissionEnum.getAllPermissionByType(1);
			}
			setSinglePolicyPermissionInRanger("hdfs", hiveMetaStoreOnHdfsDir, hdfsPathPermissions, users);
			setSinglePolicyPermissionInRanger("hdfs", hiveMetaStoreExternalOnHdfsDir, hdfsPathPermissions, users);
		}else {
			setSinglePolicyPermissionInRanger(type, resource, permissions, users);
		}
	}


	/**
	 * 简单增加策略，大部分参数都已写死
	 * 增加单个策略
	 * @param type
	 * @param resource
	 * @param permissions 权限字符串，使用"|"分隔
	 * @param users 用户字符串，使用"|"分隔
	 */
	public void setSinglePolicyPermissionInRanger(String type,String resource,String permissions,String users) {
		try {
			if (StringUtils.isEmpty(permissions)) {
				logger.info("setPermission:type["+type+"] resource["+resource+"] permissions["+permissions+"] users["+users+"] for delete.");
				setPermissionInRangerForDeleteOrUpdate(type,resource,users);
			} else {
				logger.info("setPermission:type["+type+"] resource["+resource+"] permissions["+permissions+"] users["+users+"] for add.");
				permissions = permissions.toLowerCase();
				setPermissionInRangerForAdd(type,resource,permissions,users);
			}
		} catch (Exception e) {
			logger.error(e.getMessage());
			throw new ApiException("RangerService setPermissionInRanger error.");
		}
	}


    /**
     * 直接引用
     *
     * @param type
     * @param resource
     * @param permissions
     * @param users
     */
    public void setComplexPermissionInRanger(String type, String resource, String permissions, String users) {
		setPermissionInRanger(type,resource,permissions,users);
    }

	/**
	 * 如果传递的权限设置中permissions为空，则表示更新或者删除操作
	 * 更新是指原有一个或多个子策略，按users更新其中子策略
	 * 如果不存在子策略了，则将策略删除
	 * @param type
	 * @param resource
	 * @param users
	 */
	private void setPermissionInRangerForDeleteOrUpdate (String type, String resource, String users) {
		//根据资源获取资源ID，看策略是否已存在。
		int policyId = newSmartAdd(type,resource);
		if (policyId < 0) return;
        //根据ID拿策略内容
        String data = rangerRestService.getPolicyById(policyId + "");
		JSONObject remoteObject = JSONObject.parseObject(data);
		//解析策略
		JSONArray updatedPolicyItems = new JSONArray();
		JSONArray remotePolicyItems = remoteObject.getJSONArray("policyItems");
        String[] userArray = users.split("\\|");
		//逐个对比
		for (int i=0; i<remotePolicyItems.size(); i++) {
			JSONObject remoteItem = remotePolicyItems.getJSONObject(i);
			String remoteUser = remoteItem.getJSONArray("users").getString(0);
			//如果远处子策略要被设置为空，则不再添加
            boolean isUsersContain = false;
            for (String user : userArray) {
                if (user.equalsIgnoreCase(remoteUser)) {
                    isUsersContain = true;
                    break;
                }
            }
            if (isUsersContain) {
				continue;
			}
			updatedPolicyItems.add(remoteItem);
		}
		//如果一条子策略也没有，则删除策略
		if (updatedPolicyItems.size() == 0) {
            try {
                rangerRestService.delPolicy(policyId + "");
			} catch (Exception e) {
				logger.error(e.getMessage());
                throw new ApiException("RangerService setPermissions for delete policy[" + policyId + "] error.");
			}
			return;
		}
		//如果还有子策略，则更新
		remoteObject.put("policyItems", updatedPolicyItems);
        try {
            doAddOrUpdateRequest(policyId, remoteObject);
		}catch (Exception e) {
			logger.error(e.getMessage());
            throw new ApiException("RangerService setPermissions for update policy[" + policyId + "] error.");
        }

	}

	/**
	 * 根据传参，且权限不为空，则为增加
	 * @param type
	 * @param resource
	 * @param permissions
	 * @param users
	 */
	private void setPermissionInRangerForAdd(String type, String resource, String permissions, String users) {
		RangerEntity rangerEntity = new RangerEntity();
		List<RangerResource> rangerResourceList = new ArrayList<>();
		List<RangerPolicyItem> rangerPolicyItemList = new ArrayList<>();
		rangerEntity.setType(type);
		rangerEntity.setResource(resource);
		//不同类型的资源值不同
		if (type.equalsIgnoreCase("hdfs")) {
			RangerResource rr = new RangerResource();
			rr.setKey("path");
			rr.addValue(resource);
			rr.setIsExcludes("false");
            rr.setIsRecursive("true");
			rangerResourceList.add(rr);
			rangerEntity.setRangerResourceList(rangerResourceList);
		} else if (type.equalsIgnoreCase("hbase")) {
			String[] resourceArray = resource.split("\\|");
			String table = resourceArray[0];
			String columnFamily = resourceArray[1];
			String column = resourceArray[2];
			RangerResource hbase_rr1 = new RangerResource();
			hbase_rr1.setKey("table");
			hbase_rr1.addValue(table);
			RangerResource hbase_rr2 = new RangerResource();
			hbase_rr2.setKey("column-family");
			hbase_rr2.addValue(columnFamily);
			RangerResource hbase_rr3 = new RangerResource();
			hbase_rr3.setKey("column");
			hbase_rr3.addValue(column);
			rangerResourceList.add(hbase_rr1);
			rangerResourceList.add(hbase_rr2);
			rangerResourceList.add(hbase_rr3);
			rangerEntity.setRangerResourceList(rangerResourceList);
        } else if (type.equalsIgnoreCase("hive")) {
            String[] resourceArray = resource.split("\\|");
            String database = resourceArray[0];
            String table = resourceArray[1];
            String column = resourceArray[2];
            RangerResource hive_rr1 = new RangerResource();
            hive_rr1.setKey("database");
            hive_rr1.addValue(database);
            RangerResource hive_rr2 = new RangerResource();
            hive_rr2.setKey("table");
            hive_rr2.addValue(table);
            RangerResource hive_rr3 = new RangerResource();
            hive_rr3.setKey("column");
            hive_rr3.addValue(column);
            rangerResourceList.add(hive_rr1);
            rangerResourceList.add(hive_rr2);
            rangerResourceList.add(hive_rr3);
            rangerEntity.setRangerResourceList(rangerResourceList);
        } else if (type.equalsIgnoreCase("yarn")) {
            RangerResource rr = new RangerResource();
            rr.setKey("queue");
            rr.addValue(resource);
            rr.setIsExcludes("false");
            rr.setIsRecursive("true");
            rangerResourceList.add(rr);
            rangerEntity.setRangerResourceList(rangerResourceList);
        } else if (type.equalsIgnoreCase("kafka")) {
            RangerResource rr = new RangerResource();
            rr.setKey("topic");
            rr.addValue(resource);
            rr.setIsExcludes("false");
            rr.setIsRecursive("false");
            rangerResourceList.add(rr);
            rangerEntity.setRangerResourceList(rangerResourceList);
        }
		//为每个用户单独创建一条子策略，方便后续更改
		String[] userArray = users.split("\\|");
		for (String user : userArray) {
			RangerPolicyItem rpi = new RangerPolicyItem();
			String[] permissionArray = permissions.split("\\|");
			for (String permission : permissionArray) {
				rpi.addAccess(permission);
			}
			rpi.addUsers(user);
			rpi.setDelegateAdmin("false");
			rangerPolicyItemList.add(rpi);
		}
		rangerEntity.setRangerPolicyItemList(rangerPolicyItemList);
		//设置策略名称（如果是hdfs，需要将路径格式转化）
		String policyName = formatPolicyName(type, resource);
		rangerEntity.setEnabled(true);
		rangerEntity.setAuditEnabled(true);
		rangerEntity.setPolicyName(policyName);
		rangerEntity.setDescription("add policy for "+type+" and resource is "+resource+"");

		addRangerEntity(rangerEntity);
	}


	/**
	 * 根据策略类型和策略资源值来删除策略
	 * @param type
	 * @param resource
	 */
	public Map<String, Object> deletePolicyByResource(String type,String resource) {
		Map<String, Object> result = new HashMap<>();
        try {
            int policyId = rangerRestService.getPolicyIdByResourceNameAndPolicyName(type, resource);
			logger.info("RangerService deletePolicyByResource policyId is "+policyId);
			if (policyId == -1) {
				result.put("code", "404");
				result.put("mesg", "所要删除的策略不存在。");
            }
            result = rangerRestService.delPolicy(policyId + "");
		}catch (Exception e) {
			logger.error(e.getMessage());
            throw new ApiException("RangerService deletePolicyByResource type[" + type + "] resource[" + resource + "] error.");
		}
		return result;
	}


	/**
	 * 根据用户名删除ranger中的用户
	 * @param name
	 * @return
	 */
	public void deleteUserByName(String name) {
		try {
            logger.info("RangerService begin to deleteUserByName :" + name);
            rangerRestService.delUser(name);
			logger.info("RangerService ends to deleteUserByName :" +name);
		} catch (Exception e) {
			logger.error(e.getMessage());
            throw new ApiException("RangerService deleteUserByName name[" + name + "] error.");
		}
	}

	/**
	 * 根据用户名删除ranger中的同名组
	 * @param name
	 * @return
	 */
	public void deleteGroupByName(String name) {
        try {
            rangerRestService.delGroup(name);
		} catch (Exception e) {
			logger.error(e.getMessage());
            throw new ApiException("RangerService deleteGroupByName name[" + name + "] error.");
		}
	}

	/**
	 * 根据配置文件获取ranger的api接口地址
	 * @return
	 */
	private String getRangeUrl() {
        Map map = clusterConfigUtil.getRangerConfig();
		return "http://" + map.get("host_name") + ":" + map.get("ranger.service.http.port");
	}

    /**
     * 根据配置文件获取ranger的hostname值
     *
     * @return
     */
    private String getRangeHostName() {
        Map map = clusterConfigUtil.getRangerConfig();
        return map.get("host_name").toString();
    }

	/**
	 * 得到一个策略的JSON对象以后
	 * 解析出其中某个policyItem的权限数组，返回字符串
	 * @param accessArray
	 * @return
	 */
	private String getAccessToString(JSONArray accessArray) {
		Iterator<Object> it = accessArray.iterator();
		String permissions = "";
		while(it.hasNext()){
			JSONObject access = (JSONObject)it.next();
			permissions += access.getString("type")+"|";
		}
		if (permissions.endsWith("|"))
			permissions = permissions.substring(0,permissions.length()-1);
		return permissions;

	}

	/**
	 * 判断远程已有的子策略中，用户是否在本地要更新的列表中
	 * @param user
     * @return
     */
    private boolean isRemoteUserExistsToUpdate(String user, List<RangerPolicyItem> rangerPolicyItems) {
		for (RangerPolicyItem item : rangerPolicyItems) {
			String localUser = item.getUsers().get(0);
			if (user.equalsIgnoreCase(localUser)) {
				return true;
			}
		}
        return false;
    }

    /**
     * 组装新的策略JSON对象
     */
    private JSONObject buildJsonObject(int policyId, Map<String, String> rangerParams, List<RangerPolicyItem> rangerPolicyItems, List<RangerResource> rangerResources) {
        JSONObject policyBody;
        if (policyId < 0) {
            policyBody = buildJsonObjectForAdd(rangerParams, rangerPolicyItems, rangerResources);
            return policyBody;
        }
        //根据id拿到已有的策略信息
        String data = rangerRestService.getPolicyById(policyId + "");
		JSONObject obj = JSONObject.parseObject(data);
		//对比JSONArray的内容
		JSONArray updatedPolicyItems = new JSONArray();
		JSONArray remotePolicyItems = obj.getJSONArray("policyItems");
		//添加远处可用的子策略
		for (int i=0; i<remotePolicyItems.size(); i++) {
			JSONObject remoteItem = remotePolicyItems.getJSONObject(i);
			JSONArray remoteUserArray = remoteItem.getJSONArray("users");
			if (remoteUserArray.isEmpty()) {
				continue;
			}
			String remoteUser = remoteUserArray.getString(0);
			//如果用户名是否存在于本地要更新的列表中，是则直接跳到下一个用户。
            if (isRemoteUserExistsToUpdate(remoteUser, rangerPolicyItems)) continue;
			//判断用户的权限列表是否为空，是则直接跳到下一个用户。
			JSONArray remoteAccess = remoteItem.getJSONArray("accesses");
			if (remoteAccess.size() == 0) continue;
			//走到这里说明子策略可用
			updatedPolicyItems.add(remoteItem);
        }
        //添加本地要修改的子策略
        for (int j = 0; j < rangerPolicyItems.size(); j++) {
			RangerPolicyItem rpi = rangerPolicyItems.get(j);
			//如果本地权限串为空，则跳过
			if (rpi.getAccess().size() == 0) continue;
			JSONObject localItem = rpi.toJsonObj();
			updatedPolicyItems.add(localItem);
		}
		obj.put("policyItems", updatedPolicyItems);
        return obj;
    }

    /**
     * 完全新增策略时，组装新的策略JSON对象
     */
    private JSONObject buildJsonObjectForAdd(Map<String, String> rangerParams, List<RangerPolicyItem> rangerPolicyItems, List<RangerResource> rangerResources) {
        JSONObject rootObj = new JSONObject();
        //填写普通键值对参数
        for (Map.Entry<String, String> entry : rangerParams.entrySet()) {
            rootObj.put(entry.getKey(), entry.getValue());
        }
        //填写policyItems
        JSONArray policyItemList = new JSONArray();
        for (RangerPolicyItem item : rangerPolicyItems) {
            policyItemList.add(item.toJsonObj());
        }
        rootObj.put("policyItems", policyItemList);
        //填写resources
        JSONObject resList = new JSONObject();
        for (RangerResource res : rangerResources) {
            resList.put(res.getKey(), res.toJsonObj());
        }
        rootObj.put("resources", resList);
        return rootObj;
    }


    /**
     * 执行ranger的同步Linux用户和组功能
     *
     * @param user
     * @param group
     */
    public void rangerUserSync(String user, String group) {
        String hostName = getRangeHostName();
        String linuxName = ambariConfig.getAmbariServerLoginUser();
//        String linuxPwd = ambariConfig.getAmbariServerLoginPass();
        int sshPort = Integer.parseInt(ambariConfig.getSshPort());
        SshUtil sshUtil;
        try {
            sshUtil = remoteIp(hostName, sshPort, linuxName, null);
        } catch (Exception e) {
            logger.info(e.getMessage());
            throw new ApiException("RangerService rangerUserSync connect remoteIp error.");
        }
        //将需要同步的用户用组写入一个文件中，因为同步的命令只支持从文件中获取用户和组。
        String fileName = "/tmp/" + user + "_" + UUID.randomUUID() + ".txt";
        String writeTxtFileCmd = " echo '\"" + user + "\",\"" + group + "\",\"bigdata\"' > " + fileName + "";
        logger.info("RangerService rangerUserSync writeTxtFileCmd:" + writeTxtFileCmd);
        try {
            sshUtil.runCmd(writeTxtFileCmd, "UTF-8");
        } catch (Exception e) {
            logger.info(e.getMessage());
            throw new ApiException("RangerService rangerUserSync run writeTextFileCmd error.");
        }
        //执行同步的执令
        String javaHome = ambariConfig.getJavaHome();
        String exeUserSyncCmd = javaHome + "/bin/java -Dlogdir=/var/log/ranger/usersync" +
                " -cp /usr/hdp/current/ranger-usersync/dist/*:" +
                "/usr/hdp/current/ranger-usersync/lib/*:" +
                "/usr/hdp/current/ranger-usersync/conf " +
                "org.apache.ranger.unixusersync.process.FileSourceUserGroupBuilder " + fileName + "";
        String deleteTxtFileCmd = "rm -f " + fileName;
        try {
            logger.info("RangerService rangerUserSync exeUserSyncCmd:" + exeUserSyncCmd);
            String result = sshUtil.runCmdWithReturnInfo(exeUserSyncCmd, "UTF-8");
            logger.info(result);
            logger.info("RangerService rangerUserSync deleteTxtFileCmd:" + deleteTxtFileCmd);
            sshUtil.runCmd(deleteTxtFileCmd, "UTF-8");
        } catch (Exception e) {
            logger.info(e.getMessage());
            throw new ApiException("RangerService rangerUserSync run exeUserSyncCmd error.");
        }
    }

	Registry<ConnectionSocketFactory> registry = RegistryBuilder.<ConnectionSocketFactory>create().register("http", PlainConnectionSocketFactory.getSocketFactory())
			.register("https", SSLConnectionSocketFactory.getSocketFactory()).build();
	RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(30000).setSocketTimeout(15000).setConnectionRequestTimeout(30000).build();


	public Map getRangeConfig(){
		Map<String,String> config=new HashMap<>();
		config.put("rangerusersync_user","rangerusersync");
		String password=clusterConfigUtil.getRangerSyncPassword();
		if(password==null ||"".equals(password)){
			logger.error("get ranger sync password is null");
		}
		config.put("rangerusersync_user_password",password);
		return config;
	}

	private OpResult doRangerRequest(String requestType, String url, Map<String, String> header, String requestBody) throws org.apache.http.HttpException, IOException {
		Map config=getRangeConfig();
		//String authorization = "Basic " + Base64Util.encode(config.get("ranger_server_user") + ":" + config.get("ranger_server_pass"));
		//rangerusersync_user 用户权限更大，而且密码没有加密，就用这个了。
		String authorization = "Basic " + Base64Util.getBase64(config.get("rangerusersync_user") + ":" + config.get("rangerusersync_user_password"));
		if(header==null){
			header=new HashMap<>();
		}
		header.put("Accept","application/json");
		if(header.get("Authorization")==null) {//没带认证，就用默认账号
			header.put("Authorization", authorization);
		}
		logger.info("request url={} ", url);
		CloseableHttpClient client = HttpClients.custom().setConnectionManager(new PoolingHttpClientConnectionManager(registry)).setDefaultRequestConfig(requestConfig).build();
		CloseableHttpResponse response = null;
		HttpEntity stringEntity=new StringEntity("");
		if (requestBody!=null && !requestType.equalsIgnoreCase("GET")) {
			stringEntity = new StringEntity(requestBody, "application/json", "UTF-8");
		}
		if (("GET").equalsIgnoreCase(requestType)) {
			HttpGet httpGet = new HttpGet(url);
			if (header != null && header.size() > 0) {
				for (Map.Entry<String, String> entry : header.entrySet()) {
					httpGet.addHeader(entry.getKey(), entry.getValue());
				}
			}
			response = client.execute(httpGet);
		} else if (("POST").equalsIgnoreCase(requestType)) {
			HttpPost httpPost = new HttpPost(url);
			if (header != null && header.size() > 0) {
				for (Map.Entry<String, String> entry : header.entrySet()) {
					httpPost.addHeader(entry.getKey(), entry.getValue());
				}
			}
			httpPost.setEntity(stringEntity);
			response = client.execute(httpPost);
		} else if (("DELETE").equalsIgnoreCase(requestType)) {
			HttpDelete httpDelete = new HttpDelete(url);
			if (header != null && header.size() > 0) {
				for (Map.Entry<String, String> entry : header.entrySet()) {
					httpDelete.addHeader(entry.getKey(), entry.getValue());
				}
			}
			response = client.execute(httpDelete);
		} else if (("PUT").equalsIgnoreCase(requestType)) {
			HttpPut httpPut = new HttpPut(url);
			if (header != null && header.size() > 0) {
				for (Map.Entry<String, String> entry : header.entrySet()) {
					httpPut.addHeader(entry.getKey(), entry.getValue());
				}
			}
			httpPut.setEntity(stringEntity);
			response = client.execute(httpPut);
		}
		OpResult opResult = new OpResult();
		Integer statusCode = response.getStatusLine()!=null?response.getStatusLine().getStatusCode():0;
		logger.info("statusCode={}", statusCode);
		opResult.setHttpStatus(statusCode);
		if (response == null || statusCode == 0) {
			opResult.setStatus(0);
			opResult.setMsg("返回信息为空");
			return opResult;
		}
		String result="";
		HttpEntity entity = response.getEntity();
		if (entity != null) {
			result = EntityUtils.toString(entity, "UTF-8");
			opResult.setResult(result);
			logger.info("result={}", result);
		} else {
			logger.error("result is empty");
		}
		if (statusCode == HttpStatus.SC_OK||statusCode == HttpStatus.SC_NO_CONTENT) {
			opResult.setStatus(statusCode);
			opResult.setMsg(result);

		} else {
			opResult.setStatus(-1);
			opResult.setMsg("请求错误，状态码=" + statusCode + ",信息=" + result);
		}
		if (entity != null){
			EntityUtils.consume(entity);//释放资源
		}
		return opResult;
	}

	/**
	 *调用Ranger接口判断Ranger用户是否存在
	 * @param userName
	 * @return
	 */
	public boolean getUserByName(String userName) {
		String url=getRangeUrl() + "/service/xusers/users?name=" + userName;
		try {
			OpResult<String> opResult=doRangerRequest("GET",url,null,null);
			String result=opResult.getResult();
			logger.info("getUserByName {}",result);
			JSONObject jo = (JSONObject)JSONValue.parse(result);
			if (jo.get("vXUsers") != null) {
				JSONArray ja = (JSONArray)jo.get("vXUsers");
				Iterator<Object> it = ja.iterator();
				while (it.hasNext()) {
					JSONObject userInfo = (JSONObject) it.next();
					if (userInfo.getString("name").equalsIgnoreCase(userName)) {
						return true;
					}
				}
			}
		} catch (Exception e) {
			logger.error(" getUserByName {} {}",userName,e.getMessage());
		}
		return false;
	}

	public boolean getGroupByName(String groupName) {
		String url=getRangeUrl() + "/service/xusers/groups?name=" + groupName;
		try {
			OpResult<String> opResult=doRangerRequest("GET",url,null,null);
			String result=opResult.getResult();
			JSONObject jo = (JSONObject)JSONValue.parse(result);
			if (jo.get("vXGroups") != null) {
				JSONArray ja = (JSONArray)jo.get("vXGroups");
				Iterator<Object> it = ja.iterator();
				while (it.hasNext()) {
					JSONObject groupInfo = (JSONObject) it.next();
					if (groupInfo.getString("name").equalsIgnoreCase(groupName)) {
						return true;
					}
				}
			}
		} catch (Exception e) {
			logger.error(" getGroupByName {} {}",groupName,e.getMessage());
		}
		return false;
	}

	public boolean isExistUser(String userName) {
		boolean result=getUserByName(userName);
		logger.info("isExistUser result={}",result);
		return result;
	}
	public boolean isExistGroup(String groupName) {
		boolean result=getGroupByName(groupName);
		logger.info("isExistUser result={}",result);
		return result;
	}

	public Boolean rangerUserAddByHttp(String user, String group) throws ApiException {
		boolean existUser=isExistUser(user);
		boolean existGroup=isExistGroup(group);//先写死
		String userPassword ="bigdata";
		Map<String,String> config= getRangeConfig();
		if(config.get("rangerusersync_user_password")==null){
			logger.error("rangerusersync_user_password is empty ");
		}
		Integer newGroupId=0;
		Map<String, String> header = new HashMap<>(2);
		String authorization = "Basic " + Base64Util.getBase64(config.get("rangerusersync_user") + ":" + config.get("rangerusersync_user_password"));
		header.put("Authorization", authorization);

		if(!existGroup) {
			String addGroupUrl = getRangeUrl() + "/service/xusers/secure/groups";//添加组
			//必须使用rangerusersync用户，使用root用户会报404错误
			String addGroupJson = "{\"name\":\"" + group + "\",\n" +
					"\t\"owner\":\"rangerusersync\",\n" +
					"\t\"groupType\":1,\n" +
					"\t\"groupSource\":1,\n" +
					"\t\"description\":\"sync by ranger\"}";
			OpResult<String> addGroupResult = new OpResult<>();
			try {
				addGroupResult = doRangerRequest("POST", addGroupUrl, header, addGroupJson);
			} catch (Exception e) {
				logger.error("rangerUserAdd {},{}", user, group, e);
			}
			if (addGroupResult.getHttpStatus() ==null||addGroupResult.getHttpStatus() != 200) {
				logger.error("添加用户组报错", addGroupResult);
				throw new ApiException("添加用户组报错");
			}
			String result = addGroupResult.getResult();
			logger.info("ranger add user info {}",result);
		}else{
			//newGroupId=getGroupIdByName(group);
		}

		if(!existUser) {
			String userSync = "{\"groupIdList\":[\"" + newGroupId + "\"],\"status\":1,\"userRoleList\":[\"ROLE_USER\"],\"name\":\"" + user + "\",\"password\":\"" + userPassword + "\",\"firstName\":\"" + user + "\",\"lastName\":\"\",\"userSource\":\"1\"}";
			String userAddUrl = getRangeUrl() + "/service/xusers/secure/users";//添加用户
			OpResult<String> addUserResult;
			try {
				addUserResult = doRangerRequest("POST", userAddUrl, header, userSync);
				if (addUserResult.getHttpStatus()==null||addUserResult.getHttpStatus() != 200) {
					logger.error("添加用户报错", addUserResult);
					throw new ApiException("添加用户报错");
				}
				return true;
			} catch (Exception  e) {
				logger.error("rangerUserAdd {},{}", user, group, e);
			}
		}
		return true;
	}


    /**
     * 远程连接ranger服务器，执行同步Linux用户
     *
     * @param hostName
     * @param port
     * @param user
     * @param pwd
     * @return
     * @throws Exception
     */
    private SshUtil remoteIp(String hostName, int port, String user, String pwd) throws Exception {
        // ssh远程连接kerberos服务器
        SshConf conf = new SshConf();
        conf.setHost(hostName);
        logger.info("--SSH remote ranger server hostname--:" + hostName);
        conf.setPort(port);
        conf.setUsername(user);
        conf.setPassword(pwd);
        SshUtil sshUtil = new SshUtil(conf); // 连接服务器
        return sshUtil;
    }


    /**
     * 请求ranger连接
     *
     * @param requestType
     * @param url
     * @param requestBody
     * @return
     * @throws HttpException
     * @throws IOException
     */
    private Map<String, String> doRangerRequest(String requestType, String url, String requestBody) throws HttpException, IOException {
        HttpClient client = new HttpClient();
        HttpMethodBase httpMethodBase = getHttpMethodBase(requestType, url);
        if (!requestType.equalsIgnoreCase("GET")) {
            RequestEntity entity = new StringRequestEntity(requestBody, "application/json", "UTF-8");
            ((EntityEnclosingMethod) httpMethodBase).setRequestEntity(entity);
        }
        logger.info("do request:" + url);
        int statusCode = client.executeMethod(httpMethodBase);
        Map<String, String> mapResult = new HashMap<String, String>();
        logger.info("get response:" + httpMethodBase.getResponseBodyAsString());
        mapResult.put("statuscode", String.valueOf(statusCode));
        mapResult.put("response", httpMethodBase.getResponseBodyAsString());

        httpMethodBase.releaseConnection();
        return mapResult;
    }

    /**
     * 根据请求类型返回请求属性
     *
     * @param requestType
     * @param url
     * @return
     */
    private HttpMethodBase getHttpMethodBase(String requestType, String url) {
        HttpMethodBase httpMethodBase = new HttpMethodBase() {
            @Override
            public String getName() {
                return null;
            }
        };
        if (requestType.equalsIgnoreCase("GET")) {
            httpMethodBase = new GetMethod(url);
        } else if (requestType.equalsIgnoreCase("POST")) {
            httpMethodBase = new PostMethod(url);
        } else if (requestType.equalsIgnoreCase("PUT")) {
            httpMethodBase = new PutMethod(url);
        } else if (requestType.equalsIgnoreCase("DELETE")) {
            httpMethodBase = new DeleteMethod(url);
        }
        String userName = ambariConfig.getRangerServerUser();
        String passwd = ambariConfig.getRangerServerPass();
        String authorization = "Basic " + Base64Util.getBase64(userName + ":" + passwd);
        httpMethodBase.setRequestHeader("Authorization", authorization);
        httpMethodBase.setDoAuthentication(true);
        httpMethodBase.setRequestHeader("Content-Type", "application/json");
        return httpMethodBase;
    }

	/**
	 * 在打开kafka-plugin之后，最大权限的策略有时会出现public组的情况
	 * 如果出现了public，则ranger对kafka则失去了控制
	 * 在初始化可调用此方法过滤public组
	 */
	public void formatKafkaAllTopicPolicy() {
    	String resource = "*";
		int policyId = newSmartAdd("kafka",resource);
		if (policyId < 0) {
			return;
		}

		String data = rangerRestService.getPolicyById(policyId + "");
		JSONObject allTopicPolicyObject = JSONObject.parseObject(data);
		//解析策略
		JSONArray updatedPolicyItems = new JSONArray();
		JSONArray allTopicPolicyPolicyItems = allTopicPolicyObject.getJSONArray("policyItems");

		//逐个对比
		for (int i=0; i<allTopicPolicyPolicyItems.size(); i++) {
			JSONObject item = allTopicPolicyPolicyItems.getJSONObject(i);
			JSONArray groupsArray = item.getJSONArray("groups");
			if (groupsArray.contains("public")) {
				groupsArray.remove("public");
			}
			updatedPolicyItems.add(item);
		}
		allTopicPolicyObject.put("policyItems", updatedPolicyItems);
		try {
			doAddOrUpdateRequest(policyId, allTopicPolicyObject);
		}catch (Exception e) {
			logger.error(e.getMessage());
			throw new ApiException("RangerService formatKafkaAllTopicPolicy for update policy[" + policyId + "] error.");
		}

	}

	/**
	 * 资源格式转化，针对HDFS
	 * 如果资源名中有/，则需要更换
	 * @param type
	 * @param resource
	 * @return
	 */
	public String formatPolicyName(String type, String resource) {
		switch (type.toLowerCase(Locale.CHINA)) {
			case "hdfs" :
				//如果是最大资源，设置为all-path
				String policyName = resource;
				if ("/*".equals(resource)) {
					return "all - path";
				}
				if (policyName.startsWith("/")) {
					policyName = policyName.substring(1);
				}
				if (policyName.endsWith("/")) {
					policyName = policyName.substring(0,policyName.length()-1);
				}
				policyName = policyName.replace("/","|");
				return policyName;
			case "hbase" :
				//如果是最大资源，设置为all - table, column-family, column
				if ("*|*|*".equals(resource) ) {
					return "all - table, column-family, column";
				}
				break;
			case "hive" :
				//如果是最大资源，设置为all - database, table, column
				if ("*|*|*".equals(resource)) {
					return "all - database, table, column";
				}
				break;
			case "yarn" :
				//如果是最大资源，设置为	all - queue
				if ("*".equals(resource)) {
					return "all - queue";
				}
				break;
			case "kafka" :
				//如果是最大资源，设置为	all - queue
				if ("*".equals(resource)) {
					return "all - topic";
				}
				break;
		}
		return resource;
	}


}
